package cn.initcap;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author initcap
 * @date Created in 8/28/19 9:20 PM.
 */
public class StreamingFlinkSpecifyingTransformationFunctionsJava {

    public static void main(String[] args) throws Exception {
        // 1. 获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 获取数据
        DataStreamSource<String> text = env.socketTextStream("localhost", 9999);

        // 3. 进行转换
        text.flatMap(new MyFlatMapFunction())
                .keyBy("word")
                .timeWindow(Time.seconds(5))
                .sum("count")
                .print()
                .setParallelism(1);

        // 4. 提交执行
        env.execute("StreamingFlinkSpecifyingTransformationFunctionsJava");
    }

}
